const { BrokerAsPromised: Broker, withDefaultConfig } = require('rascal');
const definitions = require('./mqConfig.js');
const axios = require('axios');
const accessLogBuild = require('./accessLogBuilder')
const {logger}=require("./logger");//日志记录
const cache = require('./cacheUtils')

/**
 * 正常消息发送
 * @param msg
 * @return {Promise<void>}
 */
async function sendNormal(msg) {
    let broker;
    try {
        broker = await Broker.create(withDefaultConfig(definitions));//withDefaultConfig附带了适用于生产和测试环境的合理默认值（针对可靠性而不是速度进行了优化）
        broker.on('error', console.error);
        // Publish a message
        const publication = await broker.publish('p1', msg);
        console.log("生产者Normal消息发送完毕");
        publication.on('error', console.error);
    } catch (err) {
        console.error(err);
    } finally {
        await broker?.shutdown();
    }
}

async function sendError(msg) {
    let broker;
    try {
        broker = await Broker.create(withDefaultConfig(definitions));//withDefaultConfig附带了适用于生产和测试环境的合理默认值（针对可靠性而不是速度进行了优化）
        broker.on('error', console.error);
        // Publish a message
        const publication = await broker.publish('p2', msg);
        console.log("生产者Error消息发送完毕");
        publication.on('error', console.error);
    } catch (err) {
        console.error(err);
    } finally {
        await broker?.shutdown();
    }
}

/**
* 具体消费过程
* @param data 消费的data
* @param type normal/error 队列的消费
* @return {Promise<unknown>}
*/
const customerRequest = (data, type = 'normal') => {
    return new Promise(async (resolve, reject) => {

        try {
            const reqParams = JSON.parse(data)
            const headers = {
                'by-service-id': 'auth-server'
            }
            const headersLog = {
                'by-service-id': 'log-server'
            }
            let CurrentUser = {}
            let userDataInfo = {}
            if (reqParams?.reqToken) {
                logger.info('有token开始请求internal/token/resolve' + reqParams?.reqToken)
                userDataInfo = await axios.get(`http://127.0.0.1:8080/internal/token/resolve?token=${reqParams?.reqToken}`, { headers }).catch(error => {

                });
                if (userDataInfo) {
                    logger.info('userDataInfo-success' + userDataInfo?.data?.user?.lenovoId)
                    CurrentUser = await axios.post(`http://127.0.0.1:8080/internal/getUserIdentity`, {
                        uid: userDataInfo?.data?.user?.lenovoId,//lenovoId
                        registerChannel: userDataInfo?.data?.user?.registerChannel//注册渠道
                    }, { headers: headersLog }).catch(error => {

                    });
                    if(!CurrentUser?.data?.data){
                        if (type === 'normal') {
                            //error队列的消费异常也入库，不存user信息
                            reject({ msg: 'log-server/internal/getUserIdentity失败' })
                            return
                        } else {
                            //normal 的catch 会记录日志，error 需要单独记录日志
                            logger.error('log-server/internal/getUserIdentity异常队列也失败')
                        }
                    }
                }else{
                    if (type === 'normal') {
                        //error队列的消费异常也入库，不存user信息
                        reject({ msg: 'internal/token/resolve normal队列失败' })
                        return
                    } else {
                        //normal 的catch 会记录日志，error 需要单独记录日志
                        logger.error('internal/token/resolve 异常队列也失败')
                    }
                }

            }
            let params = {
                logId: reqParams.logId
            }
            const { loginDevice } = reqParams
            if (loginDevice && loginDevice.includes('wxapp')) {
                params.channel = '小程序'
            } else if (loginDevice && loginDevice.includes('paas')) {
                params.channel = 'PC客户端'
            }
            const uaObj = accessLogBuild.buildUA(reqParams.userAgent)
            const ipObj = await accessLogBuild.buildIP(reqParams.reqIp).catch(error => {

            })
            if(!ipObj){
                if (type === 'normal') {
                    //error队列的消费异常也入库，不存ip信息
                    reject({ msg: 'accessLogBuilder/buildIP normal队列失败' })
                    return
                } else {
                    //normal 的catch 会记录日志，error 需要单独记录日志
                    logger.error('accessLogBuilder/buildIP-异常队列也失败')
                }
            }
            let pathRes = null
            let pathList = []
            const cacheResult = await cache.getCache('pathObj')
            if (cacheResult) {
                pathList = cacheResult
            } else {
                pathRes = await axios.post(`http://127.0.0.1:8080/internal/accessPath/query`, {}, { headers: headersLog }).catch(error => {

                });
                if (pathRes?.data?.code === 200 && pathRes?.data?.data) {
                    pathList = pathRes?.data?.data
                    cache.setCache('pathObj', pathList)
                }else{
                    if (type === 'normal') {
                        //error队列的消费异常也入库，不存accessPath信息
                        reject({ msg: 'log-server/internal/accessPath/query normal失败' })
                        return
                    } else {
                        //normal 的catch 会记录日志，error 需要单独记录日志
                        logger.error('log-server/internal/accessPath/query-异常队列也获取失败')
                    }
                }
            }
            const otherObj = accessLogBuild.buildOther(reqParams, pathList || [])
            const userObj = accessLogBuild.buildUser(CurrentUser?.data?.data || {}, userDataInfo?.data?.user?.lenovoId)
            const newParams = {
                ...userObj,
                ...params,
                ...uaObj,
                ...ipObj,
                ...otherObj
            }
            const accessLogRes = await axios.post(`http://127.0.0.1:8080/internal/accessLog`, {
                ...newParams
            }, { headers: headersLog }).catch(error => {
                logger.error({
                    msg:'埋点保存接口失败',
                    err:error
                })
            });
            logger.info('accessLogRes-success=>'+type+'队列', accessLogRes?.data?.data)
            resolve(accessLogRes?.data?.data)
            // 创建一个新的 accessLog 数据对象
            // const newAccessLog = new accessLog(newParams);
            // // 保存数据到数据库
            // newAccessLog.save().then(savedAccessLog => {
            // console.log('Access Log saved:', savedAccessLog);
            // }).catch(error => {
            // console.log('error')
            // errLog(error,500,'埋点数据入库失败')
            // });

        } catch (e) {
            logger.error({
                err: e?.error || e,
                msg: e?.msg || e,
            })
            reject('error')
        }
    });
}

/**
 * 正常队列的消费者
 * @param i
 * @return {Promise<void>}
 */
async function normalConsumer(i) {
    try {
        const broker = await Broker.create(withDefaultConfig(definitions));//withDefaultConfig附带了适用于生产和测试环境的合理默认值（针对可靠性而不是速度进行了优化）
        broker.on('error', error => {
            logger.error(' broker Error+正常队列' + error)
        });

        // Consume a message
        const subscription = await broker.subscribe('s1'); //subscription 不存在会抛出异常
        subscription
            .on('message', async (message, content, ackOrNack) => {
                logger.messageId({
                    consumer: `normal消费者:${i}`,
                    messageId: message.properties.messageId
                })
                await customerRequest(content).catch(e=>{
                    //把消息发送到异常队列，finally标记已经消费
                    sendError(content)
                    logger.sendToErrorQueue({
                        consumer: `normal消费者:${i}`,
                        messageId: message.properties.messageId,
                        content
                    })
                }).finally(()=>{
                    ackOrNack();
                })
            })
            .on('error', error => {
                logger.error('正常队列+subscribe Error' +    error )
            })
            .on('invalid_content', (err, message, ackOrNack) => { //若无法解析内容（例如，消息的内容类型为“application/json”，但内容不是json），它将发出“invalid_content”事件
                logger.error('正常队列+Invalid content'+ err)
                ackOrNack(err);//默认nack 策略
            })
            .on('redeliveries_exceeded', (err, message, ackOrNack) => { //如果重新传递的数量超过订阅者限制，订阅者将发出“redelivery_exceeded”事件，并且可以由您的应用程序处理
                logger.error('正常队列+Redeliveries exceeded重新投递'+ err)
                ackOrNack(err, [{ strategy: 'republish', defer: 1000, attempts: 10 }, { strategy: 'nack' }]); //将消息重新发布回其来自的队列。 当指定尝试次数时，始终应该链接一个后备策略，否则如果超出尝试次数，您的消息将不会被确认或拒绝
            });
    } catch (err) {
        logger.error({
            err: err,
            msg: '正常队列消费端启动失败-其他Error',
        })
    }
    // console.log(`消费端${i}启动成功`)
}

/**
 * 异常队列的消费者
 * @param i
 * @return {Promise<void>}
 */
async function errorConsumer(i) {
    try {
        const broker = await Broker.create(withDefaultConfig(definitions));//withDefaultConfig附带了适用于生产和测试环境的合理默认值（针对可靠性而不是速度进行了优化）
        broker.on('error', error => {
            logger.error(' broker Error+异常队列' + error)
        });

        // Consume a message
        const subscription = await broker.subscribe('s2'); //subscription 不存在会抛出异常
        subscription
            .on('message', async (message, content, ackOrNack) => {
                logger.messageId({
                    consumer: `error消费者:${i}`,
                    messageId: message.properties.messageId
                })
                await customerRequest(content,'error').catch(e=>{
                    logger.error({
                        err: e,
                        msg: '异常消费队列消费也失败了',
                        messageId:message?.properties?.messageId
                    })
                }).finally(()=>{
                    ackOrNack();
                })
            })
            .on('error', error => {
                logger.error('异常队列+subscribe Error' +    error )
            })
            .on('invalid_content', (err, message, ackOrNack) => { //若无法解析内容（例如，消息的内容类型为“application/json”，但内容不是json），它将发出“invalid_content”事件
                logger.error('异常队列+Invalid content'+ err)
                ackOrNack(err);//默认nack 策略
            })
            .on('redeliveries_exceeded', (err, message, ackOrNack) => { //如果重新传递的数量超过订阅者限制，订阅者将发出“redelivery_exceeded”事件，并且可以由您的应用程序处理
                logger.error('异常队列+Redeliveries exceeded重新投递'+ err)

                ackOrNack(err, [{ strategy: 'republish', defer: 1000, attempts: 10 }, { strategy: 'nack' }]); //将消息重新发布回其来自的队列。 当指定尝试次数时，始终应该链接一个后备策略，否则如果超出尝试次数，您的消息将不会被确认或拒绝
            });
    } catch (err) {
        logger.error({
            err: err,
            msg: '异常队列消费端启动失败-其他Error',
        })
    }
}



for (let i = 0; i <= 80; i++) {
    normalConsumer(i)
}

for (let i = 0; i <= 20; i++) {
    errorConsumer(i)
}

module.exports = {
    send:sendNormal
}
